package com.xl.flinkdemo.processAPI;

import com.xl.flinkdemo.entity.SensorReading;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @ClassName KeyedProcessFunction
 * @Description TODO
 * @Author hxl
 * @Date 2021/5/24 17:17
 * Version 1.0
 **/
public class MyKeyedProcessFunction {
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);

    ParameterTool parameterTool = ParameterTool.fromArgs(args);
    String host = parameterTool.get("hostname");
    int port = parameterTool.getInt("port");



    // 从socket文本流读取数据
    // nc -lk 7777(linux自带的Navicat工具)
    DataStream<String> inputdataStream = env.socketTextStream(host, port);


    DataStream<SensorReading> resultStream = inputdataStream.map(line -> {
      String[] fields = line.split(",");
      return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
    });

    resultStream.keyBy("id").process(new MyProcessFunction()).print();

    env.execute();
  }

  //实现自定义的处理函数
  public static class MyProcessFunction extends KeyedProcessFunction<Tuple, SensorReading, Integer> {

    ValueState<Long> tsTimer;
    @Override
    public void open(Configuration parameters) throws Exception {
      tsTimer = getRuntimeContext().getState(new ValueStateDescriptor<Long>("ts-timer",Long.class));
    }

    @Override
    public void processElement(SensorReading value, Context ctx, Collector<Integer> out) throws Exception {
      out.collect(value.getId().length());

      //context
      ctx.timestamp();
      ctx.getCurrentKey();//
      //ctx.output(new OutputTag<Integer>); //侧输出流
      ctx.timerService();//定时服务
      ctx.timerService().currentWatermark();//获取当前的事件时间
      ctx.timerService().currentProcessingTime();//获取当前的处理时间

      //注册闹钟
      ctx.timerService().registerEventTimeTimer(10000L);//注册事件时间的定时器 从1970年的某个时间过了1万毫秒触发
      ctx.timerService().registerProcessingTimeTimer(10000L);//注册处理时间的定时器


      //一般这么使用
      ctx.timerService().registerEventTimeTimer(ctx.timestamp() + 1000L);//获取当前时间再过1千毫秒触发
      tsTimer.update(ctx.timestamp() + 1000L);

      //取消闹钟
      ctx.timerService().deleteEventTimeTimer(1000L);//取消定时器
      //取消定时器，从状态里拿，才能真正删除得到那个时刻的闹钟
      ctx.timerService().deleteEventTimeTimer(tsTimer.value());
    }

    //闹钟响的时候
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception {
      System.out.println(timestamp + "定时器触发");
    }
  }
}
